/**
 * Copyright © 2011 Creativor Studio.
 * About license information, please see LICENSE.txt.
 */
package org.creativor.rayson.transport.client;

import java.io.ByteArrayOutputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.logging.Logger;
import org.creativor.rayson.annotation.ClientVersion;
import org.creativor.rayson.annotation.TransferCode;
import org.creativor.rayson.api.TransferArgument;
import org.creativor.rayson.api.TransferService;
import org.creativor.rayson.api.TransferSocket;
import org.creativor.rayson.client.Rayson;
import org.creativor.rayson.common.Stream;
import org.creativor.rayson.exception.IllegalServiceException;
import org.creativor.rayson.exception.ServiceNotFoundException;
import org.creativor.rayson.exception.UnsupportedVersionException;
import org.creativor.rayson.transport.api.TimeLimitConnection;
import org.creativor.rayson.transport.common.ConnectionProtocol;
import org.creativor.rayson.transport.common.ConnectionState;
import org.creativor.rayson.transport.common.ProtocolType;
import org.creativor.rayson.transport.stream.TransferResponse;
import org.creativor.rayson.util.Log;

// TODO: Auto-generated Javadoc
/**
 * A stream base client connection.
 * 
 * @author Nick Zhang
 */
class ClientStreamConnection extends TimeLimitConnection
{
	private long id;
	private SocketAddress serverAddress;
	private static final byte VERSION = Rayson.getClientVersion();
	private static final long TIME_OUT_INTERVAL = 60 * 1000;
	private static final int BUFFER_SIZE = 1024;
	private short transferCode;
	private short clientVersion;
	private SocketChannel socketChannel;
	private ByteBuffer connectHeaderBuffer;
	private ByteBuffer connectResponseBuffer;
	private ByteBuffer transferResponseBuffer;
	private AtomicBoolean closed;
	private ConnectionManager connectionManager;
	private TransferArgument argument;
	private static Logger LOGGER = Log.getLogger();
	
	/**
	 * Instantiates a new client stream connection.
	 * 
	 * @param serverAddress
	 *            the server address
	 * @param argument
	 *            the argument
	 * @param connectionManager
	 *            the connection manager
	 * @throws IllegalServiceException
	 *             the illegal service exception
	 */
	public ClientStreamConnection(SocketAddress serverAddress, TransferArgument argument,
			ConnectionManager connectionManager) throws IllegalServiceException
	{
		this.id = ConnectionManager.getNextConnectionId();
		this.connectionManager = connectionManager;
		this.serverAddress = serverAddress;
		ClientVersion clientVersionAnnotation =
				argument.getClass().getAnnotation(ClientVersion.class);
		if (clientVersionAnnotation == null)
			clientVersion = ClientVersion.DEFAULT_VALUE;
		else
			clientVersion = clientVersionAnnotation.value();
		TransferCode code = argument.getClass().getAnnotation(TransferCode.class);
		// Verify transfer code.
		if (code == null)
			throw new IllegalServiceException("No transfer code annotation"
					+ " found in argument class");
		this.transferCode = code.value();
		this.argument = argument;
		connectHeaderBuffer = ByteBuffer.allocate(ConnectionProtocol.HEADER_LENGTH);
		connectResponseBuffer = ByteBuffer.allocate(ConnectionProtocol.RESPONSE_LENGTH);
		transferResponseBuffer = ByteBuffer.allocate(ConnectionProtocol.TRANSFER_RESPONSE_LENGTH);
		this.connectHeaderBuffer.put(getProtocol().getType());
		this.connectHeaderBuffer.put(VERSION);
		this.connectHeaderBuffer.clear();
		closed = new AtomicBoolean(false);
	}
	
	/**
	 * Inits the.
	 * 
	 * @throws IOException
	 *             Signals that an I/O exception has occurred.
	 * @throws ConnectException
	 *             the connect exception
	 * @throws ServiceNotFoundException
	 *             the service not found exception
	 * @throws UnsupportedVersionException
	 *             the unsupported version exception
	 */
	public void init() throws IOException, ConnectException, ServiceNotFoundException,
			UnsupportedVersionException
	{
		// do connect to remote server.
		SocketChannel channel = SocketChannel.open(this.serverAddress);
		this.socketChannel = channel;
		try
		{
			// write connection header to remote server in blocked mode
			this.socketChannel.write(connectHeaderBuffer);
			// read response in blocked mode.
			this.socketChannel.read(connectResponseBuffer);
			connectResponseBuffer.flip();
			ConnectionState state = ConnectionState.valueOf(connectResponseBuffer.get());
			if (state != ConnectionState.OK)
				throw new ConnectException("Get wrong connection state: " + state.name());
			socketChannel.configureBlocking(true);
			connectHeaderBuffer.clear();
			// send client version to remote.
			connectHeaderBuffer.putShort(clientVersion);
			// send transfer number to remote
			connectHeaderBuffer.putShort(transferCode);
			connectHeaderBuffer.flip();
			this.socketChannel.write(connectHeaderBuffer);
			
			// then write argument to remote server.
			ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream(BUFFER_SIZE);
			DataOutputStream dataOutputStream = new DataOutputStream(byteArrayOutputStream);
			Stream.write(dataOutputStream, argument);
			byte[] argumentData = byteArrayOutputStream.toByteArray();
			ByteBuffer argumentBuffer = ByteBuffer.allocate(2 + argumentData.length);
			argumentBuffer.putShort((short) argumentData.length);
			argumentBuffer.put(argumentData);
			argumentBuffer.flip();
			socketChannel.write(argumentBuffer);
			// read response from remote
			this.socketChannel.read(transferResponseBuffer);
			transferResponseBuffer.flip();
			TransferResponse response = TransferResponse.valueOf(transferResponseBuffer.get());
			switch (response)
			{
				case NO_ACTIVITY_FOUND:
					throw new ServiceNotFoundException("No transfer " + transferCode
							+ " service found in servder:" + response.name());
				case UNSUPPORTED_VERSION:
					throw new UnsupportedVersionException("Client version is unsuppported");
				case UNKNOWN:
					throw new IOException("Got unknown response");
				default:
					break;
			}
		} catch (IOException e)
		{
			this.socketChannel.close();
			throw e;
		} catch (ServiceNotFoundException e)
		{
			this.socketChannel.close();
			throw e;
		} catch (UnsupportedVersionException e)
		{
			this.socketChannel.close();
			throw e;
		}
		LOGGER.info(this.toString() + " builded");
	}
	
	/**
	 * Close.
	 * 
	 * @throws IOException
	 *             Signals that an I/O exception has occurred.
	 */
	@Override
	public void close() throws IOException
	{
		if (closed.compareAndSet(false, true))
		{
			this.socketChannel.close();
		}
	}
	
	/**
	 * Removes the.
	 */
	void remove()
	{
		this.connectionManager.remove(this);
	}
	
	/**
	 * Gets the id.
	 * 
	 * @return the id
	 */
	@Override
	public long getId()
	{
		return id;
	}
	
	/**
	 * Gets the protocol.
	 * 
	 * @return the protocol
	 */
	@Override
	public ProtocolType getProtocol()
	{
		return ProtocolType.TRANSFER;
	}
	
	/**
	 * This method will just return 0. All the data will be read in
	 * {@link TransferService#process(TransferArgument, TransferSocket)}.
	 * 
	 * @throws IOException
	 *             This exception will never be thrown.
	 */
	@Override
	public int read() throws IOException
	{
		return 0;
	}
	
	/**
	 * Gets the version.
	 * 
	 * @return the version
	 */
	public byte getVersion()
	{
		return VERSION;
	}
	
	/**
	 * This method will just return. All the data will be written in
	 * {@link TransferService#process(TransferArgument, TransferSocket)}.
	 * 
	 * @throws IOException
	 *             Nerver throws.
	 */
	@Override
	public void write() throws IOException
	{
		return;
	}
	
	/**
	 * Gets the transfer.
	 * 
	 * @return the transfer
	 */
	public short getTransfer()
	{
		return transferCode;
	}
	
	/**
	 * Gets the timeout interval.
	 * 
	 * @return the timeout interval
	 */
	@Override
	protected long getTimeoutInterval()
	{
		return TIME_OUT_INTERVAL;
	}
	
	/**
	 * Creates the transfer socket.
	 * 
	 * @return the transfer socket
	 * @throws IOException
	 *             Signals that an I/O exception has occurred.
	 */
	TransferSocket createTransferSocket() throws IOException
	{
		return new TransferSocketImpl(this, socketChannel.socket(), transferCode, VERSION);
	}
	
	/**
	 * To string.
	 * 
	 * @return the string
	 */
	@Override
	public String toString()
	{
		StringBuffer sb = new StringBuffer();
		sb.append("{");
		sb.append("id: ");
		sb.append(this.id);
		sb.append(", protocol: ");
		sb.append(this.getProtocol());
		sb.append(", version: ");
		sb.append(this.getVersion());
		sb.append(", last contact: ");
		sb.append(getLastContactTime());
		sb.append(", address: ");
		sb.append(this.socketChannel.socket().toString());
		sb.append("}");
		return sb.toString();
	}
}
